package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class AppLogSplitProcessFunction extends ProcessFunction<String,String> {
    //侧边流输出标记
    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;
    public AppLogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag,
                                      OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

    @Override
    public void processElement(String s,Context context, Collector<String> collector) throws Exception {
        //解析json数据
        JSONObject jsonObject = JSON.parseObject(s);
        //错误日志
        if(jsonObject.get("err") != null){
            //侧边流输出
            context.output(errorTag,s);
            //错误日志已经单独保存,无需重新保存,直接删除
            jsonObject.remove("err");
        }
        //启动日志
        if (jsonObject.get("start") != null){
            context.output(startTag,jsonObject.toJSONString());
        }
        //页面日志
        else {
            //获取曝光和动作
            JSONArray displaysJsonArray = jsonObject.getJSONArray("display");
            JSONArray actionsJsonArray = jsonObject.getJSONArray("actions");
            //删除曝光和动作
            jsonObject.remove("display");
            jsonObject.remove("actions");
            //输出页面日志
            collector.collect(jsonObject.toJSONString());
            //曝光日志
            if(displaysJsonArray != null && displaysJsonArray.isEmpty()){
                for(int i = 0;i < displaysJsonArray.size();i++){
                    JSONArray displayJsonObject = displaysJsonArray.getJSONArray(i);
                    //添加字段值
                    jsonObject.put("display",displayJsonObject);
                    //输出
                    context.output(displayTag,jsonObject.toJSONString());
                }
            }
            //动作日志
            if(actionsJsonArray != null && actionsJsonArray.isEmpty()){
                for(int i = 0;i < actionsJsonArray.size();i++){
                    JSONArray actionsJsonObject = actionsJsonArray.getJSONArray(i);
                    //添加字段值
                    jsonObject.put("action",actionsJsonObject);
                    //输出
                    context.output(actionTag,jsonObject.toJSONString());
                }
            }
        }
    }
}